package com.sdy.mq.base;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

@Slf4j
public class BaseOrderedMessageListener implements MessageListenerOrderly {
    private BaseMessageConsumer messageConsumer;
    
    public BaseOrderedMessageListener(BaseMessageConsumer messageConsumer) {
        this.messageConsumer = messageConsumer;
    }
    
    @Override
    public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
        try {
            for (MessageExt message : msgs) {
                log.debug("Receive message: Topic=[{}],Tags=[{}],MsgId=[{}],Keys=[{}]", message.getTopic(), message.getTags(), message.getMsgId(), message.getKeys());
                if (!messageConsumer.consume(message)) {
                    log.error("Consumer messages failed, reconsume later.");
                    return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
                }
            }
        } catch (Exception e) {
            log.error("Consumer messages exception, reconsume later.", e);
            return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
        }
        return ConsumeOrderlyStatus.SUCCESS;
    }
}
